use std::future::Future;
use std::sync::Arc;
use std::time::Duration;

use tokio::net::{TcpListener, TcpStream};
use tokio::sync::{broadcast, mpsc, Semaphore};
use tokio::time;
use tracing::{debug, error, info, instrument};

use crate::{Command, Connection, Db, DbDropGuard, Shutdown};

const MAX_CONNECTIONS: usize = 250;

/// 运行mini-redis服务器
///
/// 接受来自提供的侦听器的连接。
/// 对于每个入站连接，生成一个任务来处理该连接。
/// 服务器将一直运行到 shutdown future 完成，此时服务器将正常关闭`
/// tokio:：signal:：ctrl_c（）`可以用作“shutdown”参数。这将监听SIGINT信号。
pub async fn run(listener: TcpListener, shutdown: impl Future) {
    //当shutdown后，通过broadcast channel 发送一个消息给所有的active connection
    let (notify_shutdown, _) = broadcast::channel(1);
    // shutdown完成后，通过mpsc channel 发送消息
    let (shutdown_complete_tx, shutdown_complete_rx) = mpsc::channel(1);
    // 连接限制信号量
    let limit_connection_semaphore = Arc::new(Semaphore::new(MAX_CONNECTIONS));

    let mut server = Listener {
        listener,
        db_holder: DbDropGuard::new(),
        limit_connections: limit_connection_semaphore,
        notify_shutdown,
        shutdown_complete_tx,
        shutdown_complete_rx,
    };

    // select! 在多个异步计算中等待，并在单个计算完成后返回
    // 同时运行服务器并侦听“关机”信号。
    // 服务器任务将一直运行，直到遇到错误，因此在正常情况下，此选项为“select！”语句运行，直到收到“shutdown”信号`！
    // 每个选项写法：
    // ```<result of async-op> = <async-op> => <step to perform with result>````
    // 所有的 `<async-op>`语句同时执行.
    tokio::select! {
        // 执行listener的run方法：TCP侦听和每个连接状态的初始化
        res = server.run() => {
            // 在此处收到错误（TcpListener的多次失败），服务器将放弃并关闭。
            // 处理单个连接时遇到的错误不会冒泡到此处
            if let Err(err) = res{
                error!(cause = %err, "failed to accept");
            }
        }
        // shutdown 信号监听
        _ = shutdown =>{
            // 监听到shutdown的信号
            info!("shutting down");
        }
    }

    // 提取shutdown_complete_rx,shutdown_complete_tx,notify_shutdown
    let Listener {
        mut shutdown_complete_rx,
        shutdown_complete_tx,
        notify_shutdown,
        ..
    } = server;

    // 显式删除 notify_shutdown
    drop(notify_shutdown);
    // 显式删除 shutdown_complete_tx
    drop(shutdown_complete_tx);
    // 等待接收shutdown完成的消息
    let _ = shutdown_complete_rx.recv().await;
}

/// 监听器
struct Listener {
    listener: TcpListener,
    db_holder: DbDropGuard,
    limit_connections: Arc<Semaphore>,
    notify_shutdown: broadcast::Sender<()>,
    shutdown_complete_tx: mpsc::Sender<()>,
    shutdown_complete_rx: mpsc::Receiver<()>,
}

impl Listener {
    async fn run(&mut self) -> crate::Result<()> {
        info!("accepting inbound connections");

        loop {
            // 获取信号量
            let permit = self.limit_connections
                .clone()
                .acquire_owned()
                .await
                .unwrap();

            // 接受一个新的 socket
            // “accept”方法在内部尝试恢复错误，因此此处的错误是不可恢复的
            let socket = self.accept().await?;

            // 每个连接创建一个handler
            let mut handler = Handler {
                db: self.db_holder.db(),
                // 连接状态：分配读写缓冲区以执行redis协议帧解析
                connection: Connection::new(socket),
                // 接收 shutdown notification
                shutdown: Shutdown::new(self.notify_shutdown.subscribe()),
                // shutdown 完成后发送消息
                _shutdown_complete: self.shutdown_complete_tx.clone(),
            };

            // 生成新任务以处理连接
            tokio::spawn(async move {
                // 处理连接
                if let Err(err) = handler.run().await {
                    error!(cause = ?err, "connection error");
                }

                // drop permit，将permit 返回给semaphore
                drop(permit);
            });
        }
    }

    /// 接受连接。
    ///
    /// 策略：每次失败后等待时间翻倍。如果等待64秒后第6次尝试失败，则返回错误。
    async fn accept(&mut self) -> crate::Result<TcpStream> {
        let mut backoff = 1;
        loop {
            match self.listener.accept().await {
                Ok((socket, _)) => return Ok(socket),
                Err(err) => {
                    if backoff > 64 {
                        // 失败次数超过6次，返回错误
                        return Err(err.into());
                    }
                }
            }

            // 暂停
            time::sleep(Duration::from_secs(backoff)).await;
            // 等待时间翻倍
            backoff *= 2;
        }
    }
}

#[derive(Debug)]
struct Handler {
    db: Db,
    connection: Connection,
    shutdown: Shutdown,
    _shutdown_complete: mpsc::Sender<()>,
}

impl Handler {
    /// 处理单个连接
    ///
    /// 从socket读取请求帧并进行处理。响应被写回socket。
    /// 当收到关闭信号时，将处理连接，直到其达到安全状态，此时将终止连接。
    #[instrument(skip(self))]
    async fn run(&mut self) -> crate::Result<()> {
        while !self.shutdown.is_shutdown() {
            // 读取请求帧时，还要监听shutdown信号
            let maybe_frame = tokio::select! {
                // 读取请求帧
                res = self.connection.read_frame() => res?,
                // 监听shutdown信号
                _ = self.shutdown.recv() =>{
                    return Ok(());
                }
            };

            // maybe_frame : 如果没有数据，为None
            let frame = match maybe_frame {
                Some(frame) => frame,
                None => return Ok(()),
            };

            // 将请求帧转换为命令
            let cmd = Command::from_frame(frame)?;

            debug!(?cmd);

            // 执行命令
            cmd.apply(&self.db, &mut self.connection, &mut self.shutdown).await?;
        }

        Ok(())
    }
}